package com.sensei.search.nodes;

import it.unimi.dsi.fastutil.ints.IntSet;

import org.apache.log4j.Logger;

import com.browseengine.bobo.api.BrowseResult;
import com.google.protobuf.TextFormat.ParseException;
import com.linkedin.norbert.NorbertException;
import com.linkedin.norbert.cluster.Node;
import com.linkedin.norbert.cluster.javaapi.Cluster;
import com.linkedin.norbert.cluster.javaapi.ClusterListenerAdapter;
import com.linkedin.norbert.cluster.javaapi.Router;
import com.linkedin.norbert.network.javaapi.NetworkClient;
import com.sensei.search.cluster.routing.UniformRouter;
import com.sensei.search.req.SenseiRequest;
import com.sensei.search.req.protobuf.SenseiRequestBPO;
import com.sensei.search.req.protobuf.SenseiRequestBPOConverter;

public class SenseiBroker extends ClusterListenerAdapter  {
	private final static Logger logger = Logger.getLogger(SenseiBroker.class);
	private final Cluster _cluster;
	
	private final NetworkClient _networkClient;
	
	private volatile UniformRouter _router = null;
	
	private static final SenseiScatterGatherHandler _scatterGatherHandler = new SenseiScatterGatherHandler();
	
	private static BrowseResult doBrowse(NetworkClient networkClient,SenseiRequest req,IntSet partitions) throws Exception{
		int size = 0;
		if (partitions!=null && (size=partitions.size())>0){
		 /* int oldOffset = req.getOffset();
		  int oldCount = req.getCount();
		  
		  req.setOffset(0);
		  req.setCount(oldCount);
		  */
		  if (req.getQuery() == null){
			  //req.setQuery("*:*");
			  req.setQuery("");
		  }
		  /*
		  if (req.getFilterIDs() == null){
			  req.setFilterIDs(null);
		  }
		  */
		  SenseiRequestBPO.Request msg = SenseiRequestBPOConverter.convert(req);
		  /*
	      List<ResponseIterator> resultFutures = new ArrayList<ResponseIterator>(size);
	      
	      List<BrowseResult> resultList = new ArrayList<BrowseResult>(size);
	      ResponseIterator it = networkClient.sendMessage(partitions.toIntArray(), msg);
    	  resultFutures.add(it);
	      
	      List<BrowseResult> boboBrowseList = new ArrayList<BrowseResult>(resultList.size());
	      
	      for (ResponseIterator resultFuture : resultFutures){
	    	  Response resp = resultFuture.next(TIMEOUT_MILLIS, TimeUnit.MILLISECONDS);
	    	  if (resp==null){
	    		  logger.error("Request Timed Out");
	    	  }
	    	  else {
	    		  if (resp.isSuccess()){
	    		    Message boboMsg = resp.getMessage();
	    		    BrowseResult res = SenseiRequestBPOConverter.convert((SenseiResultBPO.Result)boboMsg);
	    		    boboBrowseList.add(res);
	    	      }
	    	      else{
	    		    logger.error("Request Failed: ",resp.getCause());
	    	      }
	    	  }
	      }
	      
	      BrowseResult res = ResultMerger.merge(req, boboBrowseList);
	      req.setOffset(oldOffset);
	      req.setCount(oldCount);*/
		  
		  int[] partToSend = req.getPartitions();
		  if (partToSend==null){
			  partToSend = partitions.toIntArray();
		  }
		  BrowseResult res;
		  if (partToSend.length>0){
		    res = networkClient.sendMessage(partitions.toIntArray(), msg, _scatterGatherHandler);
		  }
		  else{
			res = new BrowseResult();  
		  }
	      return res;
		}
		else{
			logger.warn("no server exist to handle request.");
			return new BrowseResult();
		}
	}
	
	public SenseiBroker(Cluster cluster,NetworkClient networkClient) throws NorbertException{
		_cluster = cluster;
		_cluster.addListener(this);
		_router = (UniformRouter)_cluster.getRouter();
		
		_networkClient = networkClient;
	}
	
	public BrowseResult browse(SenseiRequest req){
		UniformRouter router = _router;
		IntSet parts = router.getPartitions();
		try {
			return doBrowse(_networkClient,req,parts);
		} catch (Exception e) {
			logger.error(e.getMessage(),e);
			return null;
		}
	}
	
	@Override
	public void handleClusterConnected(Node[] nodes, Router router) {
		_router = (UniformRouter)router;
	}


	@Override
	public void handleClusterNodesChanged(Node[] nodes, Router router) {
		_router = (UniformRouter)router;
	}
	
	public void shutdown(){
	}
}
